package com.example.rxbyhand;

import com.example.rxbyhand.scheduler.Schedulers;

/**
 * Author by lp,on 2021/8/16/016,17:01.
 */
public class RxTest {
    public static void main(String[] args) {
        Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(Emitter<Object> emitter) {
                System.out.println("subscribe thread :" + Thread.currentThread());
                emitter.onNext("test");
//                emitter.onError(new Throwable());
                emitter.onComplete();
            }
        })
//                .map(new Function<Object, Object>() {
//            @Override
//            public Object apply(Object o) {
//                return "map:"+o;
//            }
//        })
//                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .flatMap(new Function<Object, ObservableSource<Object>>() {
                    @Override
                    public ObservableSource<Object> apply(final Object o) {
                        return Observable.create(new ObservableOnSubscribe<Object>() {
                            @Override
                            public void subscribe(Emitter<Object> emitter) {
                                System.out.println("flatMap thread :" + Thread.currentThread());
                                emitter.onNext("flatMap  " + o.toString());
                            }
                        });
                    }
                })
                .subscribe(new Observer() {
                    @Override
                    public void onSubscribe() {
                        System.out.println("onSubscribe thread :" + Thread.currentThread());
                    }

                    @Override
                    public void onNext(Object o) {
                        System.out.println("onNext thread :" + Thread.currentThread());
                        System.out.println("onNext " + o);

                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete ");
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        System.out.println("onError ");
                    }
                });

    }
}
